package com.bff.gaia.streaming.api.xjoin.client.hbase.async;

import com.bff.gaia.api.java.typeutils.RowTypeInfo;
import com.bff.gaia.configuration.Configuration;
import com.bff.gaia.streaming.api.functions.async.ResultFuture;
import com.bff.gaia.streaming.api.xjoin.client.hbase.async.rowkeydealer.AbsRowKeyModeDealer;
import com.bff.gaia.streaming.api.xjoin.client.hbase.async.rowkeydealer.QualifierQueryModeDealer;
import com.bff.gaia.streaming.api.xjoin.client.hbase.async.rowkeydealer.RowKeyEqualModeDealer;
import com.bff.gaia.streaming.api.xjoin.client.hbase.core.table.HbaseSideTableInfo;
import com.bff.gaia.streaming.api.xjoin.core.enums.ECacheContentType;
import com.bff.gaia.streaming.api.xjoin.core.side.AsyncReqRow;
import com.bff.gaia.streaming.api.xjoin.core.side.FieldInfo;
import com.bff.gaia.streaming.api.xjoin.core.side.JoinInfo;
import com.bff.gaia.streaming.api.xjoin.core.side.SideTableInfo;
import com.bff.gaia.streaming.api.xjoin.core.side.cache.CacheObj;
import com.bff.gaia.shaded.guava18.com.google.common.collect.Maps;
import com.bff.gaia.types.Row;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.TableNotFoundException;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.apache.hadoop.hbase.client.HTable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.sql.Timestamp;
import java.util.Collections;
import java.util.List;
import java.util.Map;

public class HbaseAsyncReqRow extends AsyncReqRow {
	private static final long serialVersionUID = 2098635104857937717L;

	private static final Logger LOG = LoggerFactory.getLogger(HbaseAsyncReqRow.class);

	//match to the rule of netty3
	private static final int DEFAULT_BOSS_THREADS = 1;

	private static final int DEFAULT_IO_THREADS = Runtime.getRuntime().availableProcessors();

	private static final int DEFAULT_POOL_SIZE = DEFAULT_IO_THREADS + DEFAULT_BOSS_THREADS;

	private transient Connection hConnection;

	protected transient HTable table;

	private transient AbsRowKeyModeDealer rowKeyMode;

	private String tableName;

	private String tableParent;

	private String[] colNames;

	private boolean enableQualifierQuery;

	public HbaseAsyncReqRow(RowTypeInfo rowTypeInfo, JoinInfo joinInfo, List<FieldInfo> outFieldInfoList, SideTableInfo sideTableInfo) {
		super(new HbaseAsyncSideInfo(rowTypeInfo, joinInfo, outFieldInfoList, sideTableInfo));

		tableName = ((HbaseSideTableInfo) sideTableInfo).getTableName();
		tableParent = ((HbaseSideTableInfo) sideTableInfo).getFamily();
		colNames = ((HbaseSideTableInfo) sideTableInfo).getColumnRealNames();
		HbaseSideTableInfo hbaseSideTableInfo = (HbaseSideTableInfo) sideTableInfo;
		this.enableQualifierQuery = hbaseSideTableInfo.isQualifierQuery();
	}


	@Override
	public void open(Configuration parameters) throws Exception {
		LOG.info("start open ...");
		SideTableInfo sideTableInfo = sideInfo.getSideTableInfo();
		HbaseSideTableInfo hbaseSideTableInfo = (HbaseSideTableInfo) sideTableInfo;

		org.apache.hadoop.conf.Configuration config = new org.apache.hadoop.conf.Configuration();
		config.set("hbase.zookeeper.quorum", hbaseSideTableInfo.getHost());

		try {
			hConnection = ConnectionFactory.createConnection(config);
			table = (HTable) hConnection.getTable(TableName.valueOf(tableName));
		} catch (TableNotFoundException tnfe) {
			LOG.error("Table '{}' not found ", tableName, tnfe);
			throw new RuntimeException("HBase table '" + tableName + "' not found.", tnfe);
		} catch (IOException ioe) {
			LOG.error("Exception while creating connection to HBase.", ioe);
			throw new RuntimeException("Cannot create connection to HBase.", ioe);
		}
		HbaseAsyncSideInfo hbaseAsyncSideInfo = (HbaseAsyncSideInfo) sideInfo;
		if (enableQualifierQuery) {
			rowKeyMode = new QualifierQueryModeDealer(hbaseAsyncSideInfo.getColRefType(), colNames, table,
				openCache(),sideInfo);
		} else {
			rowKeyMode = new RowKeyEqualModeDealer(hbaseAsyncSideInfo.getColRefType(), colNames, table,
				openCache(),sideInfo);
		}
		LOG.info("end open.");
	}

	@Override
	public void asyncInvoke(Row input, ResultFuture<Row> resultFuture) throws Exception {
		String rowKeyStr;
		if(enableQualifierQuery){
			rowKeyStr = buildKey(input);
		}else {
			Map<String, Object> refData = Maps.newHashMap();
			for (int i = 0; i < sideInfo.getEqualValIndex().size(); i++) {
				Integer conValIndex = sideInfo.getEqualValIndex().get(i);
				Object equalObj = input.getField(conValIndex);
				if (equalObj == null) {
					resultFuture.complete(null);
				}

				refData.put(sideInfo.getEqualFieldList().get(i), equalObj);
			}
			rowKeyStr = ((HbaseAsyncSideInfo) sideInfo).getRowKeyBuilder().getRowKey(refData);
		}

		//get from cache
		if (openCache()) {
			System.out.println("cache opened");
			CacheObj val = getFromCache(rowKeyStr);
			if (val != null) {
				if (ECacheContentType.MissVal == val.getType()) {
					dealMissKey(input, resultFuture);
					return;
				} else if (ECacheContentType.SingleLine == val.getType()) {
					Row row = fillData(input, val);
					resultFuture.complete(Collections.singleton(Row.project(row, sideInfo.getOutFieldIndexs())));
				} else if (ECacheContentType.MultiLine == val.getType()) {
					for (Object one : (List) val.getContent()) {
						Row row = fillData(input, one);
						resultFuture.complete(Collections.singleton(Row.project(row, sideInfo.getOutFieldIndexs())));
					}
				}
				return;
			}
		}

		rowKeyMode.asyncGetData(tableName, rowKeyStr, input, resultFuture, sideInfo.getSideCache());
	}

	@Override
	public Row fillData(Row input, Object sideInput) {

		List<Object> sideInputList = (List<Object>) sideInput;
		Row row = new Row(sideInfo.getOutFieldInfoList().size());
		for (Map.Entry<Integer, Integer> entry : sideInfo.getInFieldIndex().entrySet()) {
			Object obj = input.getField(entry.getValue());
//			boolean isTimeIndicatorTypeInfo = TimeIndicatorTypeInfo.class.isAssignableFrom(sideInfo.getRowTypeInfo().getTypeAt(entry.getValue()).getClass());
//			boolean isTimeIndicatorTypeInfo = false;
//			if (obj instanceof Timestamp && isTimeIndicatorTypeInfo) {
//				obj = ((Timestamp) obj).getTime();
//			}

			row.setField(entry.getKey(), obj);
		}

		for (Map.Entry<Integer, Integer> entry : sideInfo.getSideFieldIndex().entrySet()) {
			if (sideInputList == null) {
				row.setField(entry.getKey(), null);
			} else {
				row.setField(entry.getKey(), sideInputList.get(entry.getValue()));
			}
		}

		return row;
	}

	@Override
	public void close() throws Exception {
		super.close();
		LOG.info("start close ...");
		if (null != table) {
			try {
				table.close();
			} catch (IOException e) {
				// ignore exception when close.
				LOG.warn("exception when close table", e);
			}
		}
		if (null != hConnection) {
			try {
				hConnection.close();
			} catch (IOException e) {
				// ignore exception when close.
				LOG.warn("exception when close connection", e);
			}
		}
		LOG.info("end close.");
	}

	//QualifierQuery缓存主键构造
	private String buildKey(Row val) {
		StringBuilder sb = new StringBuilder("");
		for (int equalField : sideInfo.getEqualValIndex()) {
			sb.append(val.getField(equalField)).append("_");
		}

		return sb.toString();
	}

	private String buildKey(Map<String, Object> val) {
		StringBuilder sb = new StringBuilder("");
		for (String equalField : sideInfo.getEqualFieldList()) {
			sb.append(val.get(tableParent + ":" + equalField)).append("_");
		}

		return sb.toString();
	}

}